//
// Created by mio on 25-4-23.
//

#include "RtmpServer.h"

#include "../Net/EventLoop.h"
#include "RtmpConnection.h"

std::shared_ptr<RtmpServer> RtmpServer::Create(EventLoop* eventloop)
{
    std::shared_ptr<RtmpServer> server(new RtmpServer(eventloop));
    return server;
}


RtmpServer::RtmpServer(EventLoop* eventloop) : TcpServer(eventloop),
    loop_(eventloop),
    event_callbacks_(10)
{
    loop_->AddTimer([this]()
    {
        std::lock_guard<std::mutex> lock(mtx_);
        for (auto iter = rtmp_sessions_.begin(); iter != rtmp_sessions_.end();)
        {
            if (iter->second->GetClients() == 0)
            {
                rtmp_sessions_.erase(iter++);
            }
            else iter++;
        }
        return true;
    }, 3000); // 3秒执行依次, 循环执行
}

RtmpServer::~RtmpServer()
{
}

void RtmpServer::SetEventCallback(const EventCallback& cb)
{
    std::lock_guard<std::mutex> lock(mtx_);
    event_callbacks_.push_back(cb);
}

void RtmpServer::AddSession(std::string stream_path)
{
    std::lock_guard<std::mutex> lock(mtx_);
    if (rtmp_sessions_.find(stream_path) == rtmp_sessions_.end())
    {
        rtmp_sessions_[stream_path] = std::make_shared<RtmpSession>();
    }
}

void RtmpServer::RemoveSession(std::string stream_path)
{
    std::lock_guard<std::mutex> lock(mtx_);
    rtmp_sessions_.erase(stream_path);
}

RtmpSessionPtr RtmpServer::GetSession(std::string stream_path)
{
    std::lock_guard<std::mutex> lock(mtx_);
    if (rtmp_sessions_.find(stream_path) == rtmp_sessions_.end())
    {
        rtmp_sessions_[stream_path] = std::make_shared<RtmpSession>();
    }
    return rtmp_sessions_[stream_path];
}

bool RtmpServer::HasPublisher(std::string stream_path)
{
    auto session = GetSession(stream_path);
    if (!session) return false;
    return (session->GetPublisher() != nullptr);
}

bool RtmpServer::HasSession(std::string stream_path)
{
    std::lock_guard<std::mutex> lock(mtx_);
    return (rtmp_sessions_.find(stream_path) != rtmp_sessions_.end());
}

void RtmpServer::NotifyEvent(std::string type, std::string stream_path)
{
    std::lock_guard<std::mutex> lock(mtx_);
    for (auto evecnt_cb : event_callbacks_)
    {
        if (evecnt_cb) evecnt_cb(type, stream_path);
    }
}

TcpConnectionPtr RtmpServer::OnConnect(int sockfd)
{
    return std::make_shared<RtmpConnection>(shared_from_this(), loop_->GetTaskScheduler().get(), sockfd);
}